package cn.lsh.kafka.producer;

import cn.lsh.kafka.db.entity.OffsetEntity;
import org.apache.kafka.clients.producer.*;

import java.time.LocalTime;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class FirstProducer {

	public static void main(String[] args) throws InterruptedException {
		Properties properties = new Properties();
		//配置broker地址
		properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node00:9092,node01:9092,node02:9092");
		//key序列化类
		properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
		//value序列化类
		properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
		//ack应答级别，0：leader不会返回ack，返回的偏移量始终为-1，不能确认数据是否保存成功；
		// 1：leader保存完消息返回ack，但此时follower可能还没有复制消息，如果这时leader挂掉，则数据丢失
		// -1或all：leader等待ISR中的follower保存消息副本后返回ack，记录不会丢失
		properties.setProperty(ProducerConfig.ACKS_CONFIG, "-1");
		//消息失败最大重试次数，ack应答级别不能设置为0
		properties.setProperty(ProducerConfig.RETRIES_CONFIG, "3");
		//批次大小，即一次发送到kafka的数据大小。producer不是调用一次send方法就发送一条消息到kafka，而是先写入缓存区，然后一次发一批消息到kafka
		properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
		//等待时间，当超过等待时间，即便缓冲区没有达到设置的大小，producer也会将消息发送到kafka
		properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1");
		//RecordAccumulator缓冲区大小，可用于缓冲等待发送至kafka的数据的内存总字节，当缓冲区被填满后，producer立即进入阻塞状态直到空闲内存被释放出来
		properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
		//设置生产者分区规则类，默认是org.apache.kafka.clients.producer.internals.RoundRobinPartitioner
		// properties.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, "cn.lsh.kafka.partitioner.MyProducerPartitioner");
		//添加拦截器，可以多个形成拦截器链，多个用逗号隔开，执行顺序就是list的顺序
		properties.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "cn.lsh.kafka.interceptor.TimeProducerInterceptor,cn.lsh.kafka.interceptor.CountProducerInterceptor");

		KafkaProducer<Integer, String> producer = new KafkaProducer<>(properties);
		for (int i = 0; i < 10; i++) {
			//构建一个ProducerRecord，需要指定topic，可以指定分区号，
			ProducerRecord<Integer, String> record = new ProducerRecord<>("ddd-1", i,"hello world");
			// producer.send(record);
			//带回调函数接收ack应答
			Future<RecordMetadata> send = producer.send(record, (recordMetadata, e) -> {
				if (e == null) {
					System.out.println("消息发送成功：偏移量=" + recordMetadata.offset() + "，分区=" + recordMetadata.partition());
				} else {
					System.out.println("消息发送失败：" + e);
				}
			});

			try {
				//get方法会阻塞当前线程，即同步发送
				RecordMetadata recordMetadata = send.get();
				System.out.println("--------消息发送成功：偏移量=" + recordMetadata.offset() + "，分区=" + recordMetadata.partition());

			} catch (ExecutionException e) {
				e.printStackTrace();
			}
		}
		System.out.println("消息已生产完：" + LocalTime.now());
		// Thread.sleep(10000);

		//close关闭资源，会关闭RecordAccumulator缓冲区，即close会把剩余的消息发送出去；并且拦截器以及分区器的close方法也是由这个close触发调用
		producer.close();
	}

}
